package com.muyu.producer;

import com.alibaba.fastjson2.JSONObject;
import com.muyu.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @ClassName: KafkaProducer
 * @Author: Zirun.Xin
 * @Date: 2025/4/27 上午10:01
 * @Version: 1.0
 * @Description: kafka生产者
 */
@Slf4j
@Component
public class KafkaProducer {

    /**
     * kafka消息驱动模板
     */
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void send(Object obj) {
        // 发送消息
        kafkaTemplate.send(Constants.TEST_TOPIC, obj).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

            /**
             * 发送失败处理
             * @param throwable 异常
             */
            @Override
            public void onFailure(Throwable throwable) {
                log.info("topic【{}】生产者 发送消息失败【{}】", Constants.TEST_TOPIC, throwable.getMessage());
            }

            /**
             * 发送成功处理
             * @param result 发送消息结果
             */
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("topic【{}】生产者 发送消息成功【{}】", Constants.TEST_TOPIC, result.getProducerRecord());
            }
        });
    }
}
